package com.t2t.top.framework.rocketmq.client.extension.impl;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.t2t.top.framework.rocketmq.client.extension.MessageConsumer;
import com.t2t.top.framework.rocketmq.client.extension.MessageExtConsumer;
import com.t2t.top.framework.rocketmq.client.extension.SerializeType;
import com.t2t.top.framework.rocketmq.utils.DiamondUtil;
import com.taobao.diamond.manager.ManagerListenerAdapter;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import java.io.Serializable;
import java.util.List;

public class MessageConsumerImpl implements InitializingBean, DisposableBean {

    Logger logger = LoggerFactory.getLogger(getClass());

    private SerializeType serializeType = SerializeType.FASTJSON;

    //begin:rocketmq消息者配置######################################
    private String consumerGroup; // 可选配置；默认值与topic一致
    private String topic;// 必需配置： rocket mq
    private String subExpression;// 必需配置：tags的表达式,如：TagA || TagC || TagD
    private String messageModel;// 请选择首播或集群: BROADCASTING,CLUSTERING
    private Integer consumeThreadMin;// 消费消息的最小线程数量
    private Integer consumeThreadMax;// 消费消息的最大线程数量
    private Integer consumeMessageBatchMaxSize;// 指消费消息的最条数
    private DefaultMQPushConsumer consumer;// 消息消费者,如OrderFulfiller
    //end:rocketmq消息者配置######################################

    private String dataId;// diamond
    private String group;// diamond
    private String namesrvAddr;//服务器地址

    private String instanceName;
    private List<MessageConsumer<Serializable>> messageConsumers;

    @Override
    public void destroy() throws Exception {
        consumer.shutdown();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        registDiamondListener();
    }

    public void restartConsumer(String configInfo) {
        this.namesrvAddr = configInfo;

        DefaultMQPushConsumer consumerNew = new DefaultMQPushConsumer(StringUtils.defaultIfBlank(consumerGroup, topic));
        if (instanceName != null) {
            consumerNew.setInstanceName(instanceName);
        }

        consumerNew.setNamesrvAddr(configInfo);
        try {
            consumerNew.subscribe(topic, subExpression);
            if (messageModel != null) {
                consumerNew.setMessageModel(MessageModel.valueOf(messageModel));
            }
            if (consumeThreadMin != null) {
                consumerNew.setConsumeThreadMin(consumeThreadMin);
            }
            if (consumeThreadMax != null) {
                consumerNew.setConsumeThreadMax(consumeThreadMax);
            }
            if (consumeMessageBatchMaxSize != null) {
                consumerNew.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
            }
            /** 新主题producer在consumer启动前发出的消息默认是接不到的,需要如下设置 */
            consumerNew.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumerNew.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        Object ms = deserializeMsg(msg.getBody());
                        for (MessageConsumer<Serializable> messageConsumer : messageConsumers) {
                            try {
                                logger.debug("rocketmq consumer instance={}, receive message:{}", messageConsumer, msg);
                                if (messageConsumer instanceof MessageExtConsumer) {
                                    MessageExtConsumer messageExtConsumer = (MessageExtConsumer) messageConsumer;
                                    messageExtConsumer.consume(msg, (Serializable) ms);
                                } else {
                                    messageConsumer.consume((Serializable) ms);
                                }
                            } catch (Exception e) {
                                logger.error("rocketmq consumer instance=" + messageConsumer + ",receive message:" + msg, e);
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                            }
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            consumerNew.start();
            logger.info("DefaultMQPushConsumer Started.");
        } catch (Exception e) {
            logger.error("rocketmq restartConsumer error", e);
            return;
        }
        /** 这里很重要：先替换老实例再关闭老实例;如果先关闭老实例,有可能它在被程序调用着 */
        DefaultMQPushConsumer oldInsetance = consumer;
        consumer = consumerNew;
        if (oldInsetance != null) {
            oldInsetance.shutdown();
        }

    }

    protected Object deserializeMsg(byte[] body) {
        //Object ms = JSON.parseObject(new String(msg.getBody()), Object.class);
        return serializeType.deserialize(body);
    }


    private void registDiamondListener() {
        DiamondUtil.registDiamondListener(new ManagerListenerAdapter() {
            @Override
            public void receiveConfigInfo(String configInfo) {
                logger.debug("group=" + group + ",dataId=" + dataId + "==>receiveConfigInfo=" + configInfo);
                restartConsumer(configInfo);
            }
        }, group, dataId);
    }

    /**
     * only for dy bean
     *
     * @param corePoolSize
     */
    public void decThreadNum(int corePoolSize) {
        consumer.getDefaultMQPushConsumerImpl().getConsumeMessageService().updateCorePoolSize(corePoolSize);
    }

    public void setConsumerGroup(String consumerGroup) {
        this.consumerGroup = consumerGroup;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public void setSubExpression(String subExpression) {
        this.subExpression = subExpression;
    }

    public void setMessageConsumers(List<MessageConsumer<Serializable>> messageConsumers) {
        this.messageConsumers = messageConsumers;
    }

    public String getMessageModel() {
        return messageModel;
    }

    public void setMessageModel(String messageModel) {
        this.messageModel = messageModel;
    }

    public void setConsumeThreadMax(Integer consumeThreadMax) {
        this.consumeThreadMax = consumeThreadMax;
    }

    public void setConsumeThreadMin(Integer consumeThreadMin) {
        this.consumeThreadMin = consumeThreadMin;
    }

    public void setConsumeMessageBatchMaxSize(Integer consumeMessageBatchMaxSize) {
        this.consumeMessageBatchMaxSize = consumeMessageBatchMaxSize;
    }

    public void setDataId(String dataId) {
        this.dataId = dataId;
    }

    public void setGroup(String group) {
        this.group = group;
    }

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }

    public String getNamesrvAddr() {
        return namesrvAddr;
    }

    public String getInstanceName() {
        return instanceName;
    }

    public void setInstanceName(String instanceName) {
        this.instanceName = instanceName;
    }

    public void setDiamondIpList(String diamondIpList) {
        logger.warn("不推荐使用这行配置,建议配置diamond配置文件ServerAddress：<property name=\"diamondIpList\" value=\"${diamond.ipList}\" />");
    }

    public void setSerializeType(String serializeType) {
        this.serializeType = SerializeType.valueOf(serializeType != null ? serializeType.toUpperCase() : "FASTJSON");
    }

    public SerializeType getSerializeType() {
        return serializeType;
    }

    public void setSerializeType(SerializeType serializeType) {
        this.serializeType = serializeType;
    }

}
